package com.liuqi.knet.remoting.transport.netty;

import com.liuqi.knet.remoting.ChannelHandler;
import com.liuqi.knet.remoting.KNetConfig;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.transport.AbstractChannel;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Netty 通道包装类
 *
 * @author liuqi
 * @date 2025/2/8 17:03
 **/
class NettyChannel extends AbstractChannel {

    /**
     * 缓存连接的通道
     */
    static final ConcurrentMap<Channel, NettyChannel> channelMap = new ConcurrentHashMap<Channel, NettyChannel>();

    final Channel channel;

    NettyChannel(Channel channel, ChannelHandler handler, KNetConfig netConfig) {
        super(handler, netConfig);
        this.channel = channel;
    }

    /**
     * 从缓存中获取Netty通道. 如果不存在则进行创建并缓存
     *
     * @param ch
     * @param handler
     * @param netConfig
     * @return
     */
    static NettyChannel getOrAddChannel(Channel ch, ChannelHandler handler, KNetConfig netConfig) {
        if (ch == null) {
            return null;
        }
        NettyChannel nc = channelMap.get(ch);
        if (nc == null) {
            NettyChannel nettyChannel = new NettyChannel(ch, handler, netConfig);
            // 设置通道建立连接时间
            nettyChannel.setAttribute(com.liuqi.knet.remoting.Channel.connectTimestampKey, System.currentTimeMillis());
            if (ch.isActive()) {
                nc = channelMap.putIfAbsent(ch, nettyChannel);
            }
            if (nc == null) {
                nc = nettyChannel;
            }
        }
        return nc;
    }

    /**
     * 如果通道不活跃了 进行移除缓存
     *
     * @param ch
     */
    static void removeChannelIfDisconnected(Channel ch) {
        if (ch != null && !ch.isActive()) {
            channelMap.remove(ch);
        }
    }

    @Override
    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) channel.remoteAddress();
    }

    @Override
    public InetSocketAddress getLocalAddress() {
        return (InetSocketAddress) channel.localAddress();
    }

    @Override
    public boolean isActive() {
        return !isClosed() && channel.isActive();
    }

    @Override
    public void send(Object message, boolean sent) throws RemotingException {
        super.send(message, sent);

        boolean success = true;
        int timeout = getTimeout();
        try {
            ChannelFuture future = channel.writeAndFlush(message);
            if (sent) {
                success = future.await(timeout);
            }
            Throwable cause = future.cause();
            if (cause != null) {
                throw cause;
            }
        } catch (Throwable e) {
            throw new RemotingException("无法发送消息: " + message + ", 远程服务端: " + getRemoteAddress() + ", 异常消息: " + e.getMessage(), e);
        }

        if (!success) {
            throw new RemotingException("无法发送消息: " + message + ", 远程服务端: " + getRemoteAddress()
                    + ", 超时限制[" + getTimeout() + "]ms");
        }
    }

    @Override
    public void close() {
        super.close();
        channel.close();
    }

    @Override
    public int hashCode() {
        return channel.hashCode();
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof NettyChannel) {
            return channel == ((NettyChannel) obj).channel;
        }
        return false;
    }
}
